#include "my_worker_processor.h"
#include "my_event_loop.h"

char *my_worker_thread_tag = "worker_thread_tag";

my_worker_processor_t           g_worker_processor;
static int64_t                  worker_thread_idx = 0;
static int                      thread_data_slot_idx = 0;

__attribute((constructor)) static void
my_g_worker_processor_init()
{
    g_worker_processor.nthreads = 0;
    g_worker_processor.datas = NULL;
    g_worker_processor.threads = NULL;
    g_worker_processor.init_hook_chain = my_hook_chain_create();
    g_worker_processor.quit_hook_chain = my_hook_chain_create();
}


static inline void
my_worker_event_loop_init(my_event_loop_data_t *loop_data)
{
    my_hook_chain_exec(g_worker_processor.init_hook_chain, loop_data);
}


static inline void
my_worker_event_loop_quit(my_event_loop_data_t *loop_data)
{
    my_hook_chain_exec(g_worker_processor.quit_hook_chain, loop_data);
}


int
my_worker_processor_thread_data_slot_alloc()
{
    if (thread_data_slot_idx >= MY_THREAD_DATA_ITEMS) {
        MY_DEF_LOG_FATAL("worker processor thread data exhaustion");
    }

    int idx = thread_data_slot_idx;
    thread_data_slot_idx++;
    return idx;
}


// hook_add must be called before my_worker_processor_start.
void
my_worker_processor_add_init_hook(my_hook_handler_pt hook_handler)
{
    my_hook_chain_register(g_worker_processor.init_hook_chain, hook_handler);
}


// hook_add must be called before my_worker_processor_start.
void
my_worker_processor_add_quit_hook(my_hook_handler_pt hook_handler)
{
    my_hook_chain_register(g_worker_processor.quit_hook_chain, hook_handler);
}


// run in main thread.
void
my_worker_processor_start(
    int nthreads,
    int64_t mem_freelist_level,
    int64_t aligned_mem_freelists_level)
{
    (void)nthreads;
    (void)mem_freelist_level;
    (void)aligned_mem_freelists_level;

    if (nthreads <= 0) {
        MY_DEF_LOG_FATAL("worker threads must be greate than 0");
    }

    my_event_loop_data_t    *datas;
    my_thread_base_t        *threads;

    datas = my_thread_calloc(
        sizeof(my_event_loop_data_t) * (size_t)nthreads);
    threads = my_thread_calloc(
        sizeof(my_thread_base_t) * (size_t)nthreads);

    char thread_name[64] = {0};

    for (int i = 0; i < nthreads; i++) {
        snprintf(thread_name, 64, "worker-%d", i);

        my_event_loop_data_init(
            datas+i,
            my_worker_event_loop_init,
            my_worker_event_loop_quit);

        my_thread_base_start(
            my_worker_thread_tag,
            threads+i,
            my_event_loop_runner,
            datas+i,
            mem_freelist_level,
            aligned_mem_freelists_level
        );

        my_thread_set_name((threads+i)->tid, thread_name);
    }

    g_worker_processor.nthreads = nthreads;
    g_worker_processor.threads = threads;
    g_worker_processor.datas = datas;
}


void
my_worker_processor_stop()
{
    for (int i = 0; i < g_worker_processor.nthreads; i++) {
        my_event_loop_quit(g_worker_processor.threads+i);
    }

    for (int i = 0; i < g_worker_processor.nthreads; i++) {
        my_thread_base_join(g_worker_processor.threads+i, NULL);
    }

    my_thread_free(g_worker_processor.datas);
    my_thread_free(g_worker_processor.threads);
}


void
my_worker_processor_post_event(my_dispatch_event_t *dispatch_event)
{
    if (g_worker_processor.nthreads == 0) {
        MY_DEF_LOG_FATAL("worker threads must be greate than 0");
    }

    worker_thread_idx = (worker_thread_idx + 1) % g_worker_processor.nthreads;
    my_event_loop_post_event(
        g_worker_processor.threads + worker_thread_idx,
        dispatch_event);
}